/*
 * Copyright (c) 2022. China Mobile (SuZhou) Software Technology Co.,Ltd. All rights reserved.
 * Lakehouse is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

package com.chinamobile.cmss.lakehouse.engine.meta.crawler

import io.circe.generic.semiauto._
import io.circe.parser.decode
import io.circe.{Decoder, Encoder, HCursor, Json}
import org.apache.commons.lang3.StringUtils

final case class CrawlerConfig(
                                jobId: String,
                                user: String,
                                metadataUrl: String,
                                source: Source,
                                target: Target,
                                tableChangeSetting: TableChangeSetting,
                                scanStrategy: ScanStrategy
                              )

object CrawlerConfig {
  def fromJson(json: String): CrawlerConfig = {
    import Decoders._
    decode[CrawlerConfig](json).toTry.get
  }

}

trait Source extends Serializable {
  def sourceType: String

  def path: String
}

final case class LocalFile(path: String) extends Source {
  override def sourceType: String = "localFile"
}

final case class OssSource(
                            ak: String,
                            sk: String,
                            endpoint: String,
                            path: String,
                            resolver: String
                          ) extends Source {
  override def sourceType: String = "oss"
}

final case class Target(tablePrefix: String, instance: String)

final case class TableChangeSetting(
                                     tableUpdatePolicy: TableUpdatePolicy,
                                     tableDeletePolicy: TableDeletePolicy
                                   )

trait TableUpdatePolicy extends Serializable

object TableUpdatePolicy {
  object Ignore extends TableUpdatePolicy

  object AddColumn extends TableUpdatePolicy

  object Rewrite extends TableUpdatePolicy
}

trait TableDeletePolicy extends Serializable

object TableDeletePolicy {
  object Ignore extends TableDeletePolicy

  object Delete extends TableDeletePolicy
}

trait ScanStrategy extends Serializable

object ScanStrategy {
  object All extends ScanStrategy

  object Sample extends ScanStrategy
}

object Encoders {
  implicit val tableChangeSetting: Encoder[TableChangeSetting] =
    Encoder.instance(s =>
      Json.obj(
        (
          "tableUpdatePolicy",
          Json.fromString(
            StringUtils
              .removeEnd(s.tableUpdatePolicy.getClass.getSimpleName, "$")
          )
        ),
        (
          "tableDeletePolicy",
          Json.fromString(
            StringUtils
              .removeEnd(s.tableDeletePolicy.getClass.getSimpleName, "$")
          )
        )
      )
    )
}

object Decoders {

  implicit val scanStrategyDecoder: Decoder[ScanStrategy] =
    (c: HCursor) => {
      for {
        v <- c.value.as[String]
      } yield v match {
        case "All" => ScanStrategy.All
        case "Sample" => ScanStrategy.Sample
      }
    }
  implicit val tableChangeSettingDecoder: Decoder[TableChangeSetting] =
    (c: HCursor) =>
      for {
        tableUpdate <- c.downField("tableUpdate").as[String]
        tableDelete <- c.downField("tableDelete").as[String]
      } yield TableChangeSetting(
        TableUpdatePolicy.Ignore,
        TableDeletePolicy.Ignore
      )
  implicit val target: Decoder[Target] = deriveDecoder
  implicit val crawlerConfigDecoder: Decoder[CrawlerConfig] = deriveDecoder
  implicit val sourceDecoder: Decoder[Source] = (c: HCursor) => {
    val sourceType = c.downField("type").as[String]
    sourceType.flatMap(tpe =>
      tpe match {
        case "oss" =>
          Decoder
            .forProduct5("ak", "sk", "endpoint", "path", "resolver")(
              OssSource.apply
            )
            .apply(c)
      }
    )
  }
}
